package net.bwie.jtp.dws.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class JtpOrderSkuAggregataDwsJob {
    public static void main(String[] args) {

        // TODO: 2025/6/3 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // TODO: 2025/6/3 输入表-input,映射到kafka消息队列
        createInputTable(tabEnv);
        // TODO: 2025/6/3 数据处理 select
        Table reportTable = handel(tabEnv);
        // TODO: 2025/6/3 输出表output 映射到CK表
        createOutputTavle(tabEnv);
        // TODO: 2025/6/3 保存数据
        saveToDoris(tabEnv, reportTable);


    }

    private static void saveToDoris(TableEnvironment tabEnv, Table reportTable) {
        // TODO: 2025/6/3 插入doris数据
        tabEnv.executeSql(
                // 插入数据到dws_trade_sku_order_report_doris_sink表中
                "INSERT INTO dws_trade_sku_order_report_doris_sink\n" +
                // 选择相关字段并进行处理
                "SELECT\n" +
                // 将开始时间格式化为年月日时分秒格式
                "    SUBSTRING(CAST(start_time AS STRING), 0, 19) AS start_time,\n" +
                // 将结束时间格式化为年月日时分秒格式
                "    SUBSTRING(CAST(end_time AS STRING), 0, 19) AS end_time,\n" +
                // 将开始时间格式化为年月日格式，作为当前日期
                "    SUBSTRING(CAST(start_time AS STRING), 0, 10) AS cur_date,\n" +
                // 选择商品相关的字段，包括商品ID、名称、SPU ID、品牌ID和名称等
                "    sku_id, sku_name, spu_id, spu_name, tm_id, tm_name,\n" +
                // 选择商品类别相关的字段，包括三级分类、二级分类、一级分类的ID和名称
                "    category3_id, category3_name, category2_id, category2_name, category1_id, category1_name,\n" +
                // 选择订单相关的字段，包括订单数量和金额
                "    order_count, order_amount\n" +
                // 从report_table中获取数据
                "FROM report_table"
        );
    }

    private static void createOutputTavle(TableEnvironment tabEnv) {
        // TODO: 2025/6/3 创建映射doris的表
        tabEnv.executeSql(
                "CREATE TABLE dws_trade_sku_order_report_doris_sink(\n" +
                        "     `start_time` STRING,\n" +
                        "     `end_time` STRING,\n" +
                        "     `cur_date` STRING,\n" +
                        "     `sku_id` STRING,\n" +
                        "     `sku_name` STRING,\n" +
                        "     `spu_id` STRING,\n" +
                        "     `spu_name` STRING,\n" +
                        "     `tm_id` STRING,\n" +
                        "     `tm_name` STRING,\n" +
                        "     `category3_id` STRING,\n" +
                        "     `category3_name` STRING,\n" +
                        "     `category2_id` STRING,\n" +
                        "     `category2_name` STRING,\n" +
                        "     `category1_id` STRING,\n" +
                        "     `category1_name` STRING,\n" +
                        "     `order_count` BIGINT,\n" +
                        "     `order_amount` DECIMAL(16, 2)\n" +
                        ") WITH (\n" +
                        // 配置 Doris 数据库连接属性
                        "    'connector' = 'doris',\n" +
                        // 指定 Doris 集群的 FE 节点地址
                        "    'fenodes' = 'node102:8030',\n" +
                        // 设置要操作的表的标识符，格式为 database.table
                        "    'table.identifier' = 'jtp_mall_report.dws_trade_sku_order_report',\n" +
                        // 配置连接数据库的用户名
                        "    'username' = 'root',\n" +
                        // 配置连接数据库的密码
                        "    'password' = '123456',\n" +
                        // 设置 Sink 端点的批处理间隔时间
                        "    'sink.batch.interval' = '10s',\n" +
                        // 设置 Sink 端点的最大重试次数
                        "    'sink.max-retries' = '3',\n" +
                        // 设置 Sink 端点的批处理大小
                        "    'sink.batch.size' = '1000'\n" +
                        ")"
        );

    }

    private static Table handel(TableEnvironment tabEnv) {
        // TODO: 2025/6/3 创建维度表skuinfo
        tabEnv.executeSql(
                // 创建表dim_sku_info_hbase_source，用于存储SKU相关信息
                "CREATE TABLE dim_sku_info_hbase_source (\n" +
                // 定义行键字段，类型为STRING
                "      row_key STRING,\n" +
                // 定义信息字段，包含多个子字段，如id、spu_id等
                "      info ROW<id STRING, spu_id STRING, price STRING, sku_name STRING, sku_desc STRING, weight STRING, tm_id STRING, category3_id STRING, sku_default_img STRING, is_sale STRING, create_time STRING>,\n" +
                // 设置主键为row_key，但不强制执行唯一性约束
                "      PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                // 配置HBase连接器版本
                "    'connector' = 'hbase-2.2',\n" +
                // 指定HBase中的表名
                "    'table-name' = 'dim_sku_info',\n" +
                // 配置ZooKeeper集群地址
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                // 启用异步查找
                "    'lookup.async' = 'true',\n" +
                // 设置查找缓存最大行数
                "    'lookup.cache.max-rows' = '10',\n" +
                // 设置查找缓存的存活时间
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")"
        );
        // TODO: 2025/6/3
        // 创建订单与SKU信息的关联表
        // 通过SQL查询将订单详情表与SKU信息表进行左连接
        Table orderJoinTable = tabEnv.sqlQuery(
                // 构建SQL查询语句以获取订单详情和商品信息
                "SELECT\n" +
                    // 选择t1表（订单详情）中的字段
                    "    t1.id,\n" +
                    "    t1.order_id,\n" +
                    "    t1.user_id,\n" +
                    "    t1.sku_id,\n" +
                    "    t1.sku_name,\n" +
                    "    t1.province_id,\n" +
                    "    t1.create_time,\n" +
                    "    t1.source_type_name,\n" +
                    "    t1.sku_num,\n" +
                    "    t1.split_total_amount,\n" +
                    // 选择t2表（商品信息）中的字段
                    "    t2.spu_id,\n" +
                    "    t2.category3_id,\n" +
                    "    t2.tm_id\n" +
                // 从订单详情表中获取数据
                "FROM dwd_order_detail_kafka_source t1\n" +
                    // 使用左连接商品信息表，并应用系统时间旅行 joins，根据t1的处理时间获取对应的商品信息
                    "         LEFT JOIN dim_sku_info_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                    // 连接条件：订单详情表的SKU ID与商品信息表的行键（SKU ID）匹配
                    "    ON t1.sku_id = t2.row_key"
        );
        // 将生成的关联表注册为临时视图，供后续查询使用
        tabEnv.createTemporaryView("order_join_table", orderJoinTable);

        Table resultTable = tabEnv.sqlQuery(
                // 选择特定格式的时间窗口起始点和结束点，商品信息，以及在1分钟时间窗口内计算订单数量和订单金额总和
                "SELECT\n" +
                        // 将时间窗口的起始点格式化为人类可读的日期时间格式
                        "    DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') AS start_time,\n" +
                        // 将时间窗口的结束点格式化为人类可读的日期时间格式
                        "    DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') AS end_time,\n" +
                        // 选择商品的唯一标识符、名称、所属SPU、三级分类ID和品牌ID
                        "    sku_id, sku_name, spu_id, category3_id, tm_id ,\n" +
                        // 计算在当前时间窗口内不同订单的数量
                        "    count(distinct order_id) AS order_count,\n" +
                        // 计算在当前时间窗口内所有订单的金额总和，转换为decimal以确保精度
                        "    sum(cast(split_total_amount AS DECIMAL(10, 2))) AS order_amount,\n" +
                        // 获取处理数据的时间戳
                        "    PROCTIME() AS proc_time\n" +
                // 从一个时间窗口表中获取数据，该表是通过对order_join_table表中的数据按照create_time字段进行1分钟的滚动窗口操作得到的
                "FROM TABLE (\n" +
                        "        TUMBLE(\n" +
                                        "                        DATA => TABLE order_join_table,\n" +
                                        // 指定用于滚动窗口的时间字段
                                        "                        TIMECOL => DESCRIPTOR(create_time),\n" +
                                        // 定义时间窗口的大小为1分钟
                                        "                        SIZE => INTERVAL '1' MINUTES\n" +
                                        "            )\n" +
                        "    )\n" +
                // 过滤条件，排除用户ID或订单来源为空的记录
                "WHERE user_id IS NOT NULL AND source_type_name IS NOT NULL\n" +
                // 按时间窗口起始点、结束点和商品信息进行分组，以便计算每个时间窗口和商品的订单数量和金额
                "GROUP BY\n" +
                        "    window_start, window_end,\n" +
                        "    sku_id, sku_name, spu_id, category3_id, tm_id"
        );
        tabEnv.createTemporaryView("result_table", resultTable);


        tabEnv.executeSql(
                // 创建表dim_spu_info_hbase_source以定义HBase数据源的结构和配置
                "CREATE TABLE dim_spu_info_hbase_source (\n" +
                // 定义行键字段，类型为STRING
                "     row_key STRING,\n" +
                // 定义信息列，包含多个字段，如id、spu_name等
                "     info ROW<id STRING, spu_name STRING, description STRING, category3_id STRING, tm_id STRING>,\n" +
                // 设置主键为row_key，但不强制执行（可能因为HBase本身不强制要求唯一性）
                "     PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                // 配置HBase连接器的版本
                "    'connector' = 'hbase-2.2',\n" +
                // 指定HBase中的表名
                "    'table-name' = 'dim_spu_info',\n" +
                // 配置ZooKeeper集群地址，用于HBase的协调服务
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                // 启用异步查找，提高查询效率
                "    'lookup.async' = 'true',\n" +
                // 配置查找缓存的最大行数，以平衡内存使用和查询性能
                "    'lookup.cache.max-rows' = '10',\n" +
                // 配置查找缓存的存活时间，以确保数据新鲜度
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")"
        );

        tabEnv.executeSql(
                // 创建表dim_base_trademark_hbase_source，用于定义与HBase中存储的商标信息对应的表结构
                "CREATE TABLE dim_base_trademark_hbase_source (\n" +
                // 定义表的列，包括row_key作为唯一标识符
                "     row_key STRING,\n" +
                // 定义info列，包含商标的详细信息，如id、商标名称和logo URL
                "     info ROW<id STRING, tm_name STRING, logo_url STRING>,\n" +
                // 设置row_key为主键，但不强制执行，以适应HBase的分布式存储特性
                "     PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH(\n" +
                // 配置HBase连接器的版本
                "    'connector' = 'hbase-2.2',\n" +
                // 指定HBase中的表名
                "    'table-name' = 'dim_base_trademark',\n" +
                // 配置ZooKeeper集群地址，用于HBase的协调服务
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                // 启用异步查询，提高查询效率
                "    'lookup.async' = 'true',\n" +
                // 配置查询缓存的最大行数，以减少对HBase的访问次数
                "    'lookup.cache.max-rows' = '10',\n" +
                // 配置查询缓存的存活时间，平衡数据新鲜度和查询性能
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")"
        );

        // 执行SQL语句创建HBase数据源表
        tabEnv.executeSql(
                "CREATE TABLE dim_base_category3_hbase_source (\n" +
                        // 定义表的列结构\n" +
                        "        row_key STRING,\n" +
                        "        info ROW<id STRING, name STRING, category2_id STRING>,\n" +
                        // 设置主键，注意这里不强制执行主键约束\n" +
                        "        PRIMARY KEY (row_key) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        // 配置HBase连接器参数\n" +
                        "    'connector' = 'hbase-2.2',\n" +
                        // 指定HBase中的表名\n" +
                        "    'table-name' = 'dim_base_category3',\n" +
                        // 配置ZooKeeper集群地址\n" +
                        "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                        // 启用异步查询\n" +
                        "    'lookup.async' = 'true',\n" +
                        // 配置查询缓存的最大行数\n" +
                        "    'lookup.cache.max-rows' = '10',\n" +
                        // 配置查询缓存的存活时间\n" +
                        "    'lookup.cache.ttl' = '1 hour'\n" +
                        ")"
                );

        tabEnv.executeSql(
                // 创建表dim_base_category2_hbase_source，用于定义从HBase获取的类别信息
                "CREATE TABLE dim_base_category2_hbase_source (\n" +
                // 定义行键字段，类型为STRING
                "          row_key STRING,\n" +
                // 定义信息字段，包含id、name和category1_id，类型为ROW
                "          info ROW<id STRING, name STRING, category1_id STRING>,\n" +
                // 设置主键为row_key，不强制执行
                "          PRIMARY KEY (row_key) NOT ENFORCED\n" +
                // 配置表的属性
                ") WITH (\n" +
                // 设置连接器类型为hbase-2.2
                "    'connector' = 'hbase-2.2',\n" +
                // 设置HBase表名为dim_base_category2
                "    'table-name' = 'dim_base_category2',\n" +
                // 设置ZooKeeper集群地址
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                // 设置异步查询模式
                "    'lookup.async' = 'true',\n" +
                // 设置查询缓存的最大行数
                "    'lookup.cache.max-rows' = '10',\n" +
                // 设置查询缓存的存活时间
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")"
        );

        tabEnv.executeSql(
                // 创建表dim_base_category1_hbase_source，用于定义从HBase获取数据的源表结构
                "CREATE TABLE dim_base_category1_hbase_source (\n" +
                // 定义表的列结构
                // row_key: 表的主键，用于唯一标识每一行数据
                // info: 包含id和name字段的复合列，用于存储具体的数据信息
                      "      row_key STRING,\n" +
                      "      info ROW<id STRING, name STRING>,\n" +
                // 设置主键为row_key，且不强制执行（因为HBase本身是基于row_key进行数据管理的）
                      "      PRIMARY KEY (row_key) NOT ENFORCED\n" +
                      ") WITH (\n" +
                // 配置表的连接器为hbase-2.2版本
                // 'table-name': 指定HBase中的表名为dim_base_category1
                // 'zookeeper.quorum': 指定HBase集群的ZooKeeper地址，用于连接HBase集群
                // 'lookup.async': 设置为true，表示异步查找，提高查询效率
                // 'lookup.cache.max-rows': 设置查询缓存的最大行数为10行
                // 'lookup.cache.ttl': 设置查询缓存的存活时间为1小时
                      "    'connector' = 'hbase-2.2',\n" +
                      "    'table-name' = 'dim_base_category1',\n" +
                      "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                      "    'lookup.async' = 'true',\n" +
                      "    'lookup.cache.max-rows' = '10',\n" +
                      "    'lookup.cache.ttl' = '1 hour'\n" +
                      ")"
        );

        Table reportTable = tabEnv.sqlQuery(
                // 构建SQL查询语句以获取所需的销售数据信息
                "SELECT\n" +
                    // 选择商品销售数据的开始和结束时间
                    "    t1.start_time,\n" +
                    "    t1.end_time,\n" +
                    // 选择商品的SKU ID和名称
                    "    t1.sku_id,\n" +
                    "    t1.sku_name,\n" +
                    // 选择商品的SPU ID，并从SPU维度表中获取对应的SPU名称
                    "    t1.spu_id,\n" +
                    "    t2.spu_name,\n" +
                    // 选择商品的品牌ID，并从品牌维度表中获取对应的品牌名称
                    "    t1.tm_id,\n" +
                    "    t3.tm_name,\n" +
                    // 选择商品的三级分类ID，并从三级分类维度表中获取对应的分类名称
                    "    t1.category3_id,\n" +
                    "    t4.name AS category3_name,\n" +
                    // 进一步从二级分类维度表中获取二级分类的名称和ID
                    "    t4.category2_id,\n" +
                    "    t5.name AS category2_name,\n" +
                    // 最终从一级分类维度表中获取一级分类的名称和ID
                    "    t5.category1_id,\n" +
                    "    t6.name AS category1_name,\n" +
                    // 选择订单数量和金额
                    "    t1.order_count,\n" +
                    "    t1.order_amount,\n" +
                    // 获取当前时间戳，用于标记数据处理的时间
                    "    UNIX_TIMESTAMP() * 1000 AS ts\n" +
                // 从结果表中获取基础销售数据
                "FROM result_table t1\n" +
                    // 左连接SPU信息维度表，根据SPU ID获取SPU名称
                    "         LEFT JOIN dim_spu_info_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t2 ON t1.spu_id = t2.row_key\n" +
                    // 左连接品牌维度表，根据品牌ID获取品牌名称
                    "    LEFT JOIN dim_base_trademark_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t3 ON t1.tm_id = t3.row_key\n" +
                    // 左连接三级分类维度表，根据三级分类ID获取分类名称
                    "    LEFT JOIN dim_base_category3_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t4 ON t1.category3_id = t4.row_key\n" +
                    // 左连接二级分类维度表，根据二级分类ID获取分类名称
                    "    LEFT JOIN dim_base_category2_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t5 ON t4.category2_id = t5.row_key\n" +
                    // 左连接一级分类维度表，根据一级分类ID获取分类名称
                    "    LEFT JOIN dim_base_category1_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t6 ON t5.category1_id = t6.row_key"
        );
        tabEnv.createTemporaryView("report_table", reportTable);
        return reportTable;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        // TODO: 2025/6/3 输入表:映射kafka队列dwd-order-detail
        tabEnv.executeSql(
                "CREATE TABLE dwd_order_detail_kafka_source\n" +
                        "(\n" +
                        "    `id`                    STRING,\n" +
                        "    `order_id`              STRING,\n" +
                        "    `user_id`               STRING,\n" +
                        "    `order_status`          STRING,\n" +
                        "    `sku_id`                STRING,\n" +
                        "    `sku_name`              STRING,\n" +
                        "    `province_id`           STRING,\n" +
                        "    `activity_id`           STRING,\n" +
                        "    `activity_rule_id`      STRING,\n" +
                        "    `coupon_id`             STRING,\n" +
                        "    `date_id`               STRING,\n" +
                        "    `create_time`           TIMESTAMP(3),\n" +
                        "    `operate_date_id`       STRING,\n" +
                        "    `operate_time`          STRING,\n" +
                        "    `source_id`             STRING,\n" +
                        "    `source_type`           STRING,\n" +
                        "    `source_type_name`      STRING,\n" +
                        "    `sku_num`               STRING,\n" +
                        "    `split_original_amount` STRING,\n" +
                        "    `split_activity_amount` STRING,\n" +
                        "    `split_coupon_amount`   STRING,\n" +
                        "    `split_total_amount`    STRING,\n" +
                        "    `row_op_ts`             STRING,\n" +
                        "    `proc_time` AS PROCTIME(),\n" +
                        "    WATERMARK               FOR create_time AS create_time - INTERVAL '0' MINUTE\n" +
                        ") WITH (\n" +
                        // 配置Kafka连接器的属性
                        "  'connector' = 'kafka',\n" +
                        // 指定Kafka主题
                        "  'topic' = 'dwd-order-detail',\n" +
                        // 设置Kafka集群的引导服务器地址
                        "  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        // 设置消费者组ID
                        "  'properties.group.id' = 'gid-dws-trade-sku-order',\n" +
                        // 配置启动模式，从最早的偏移量开始消费
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        // 设置数据格式为JSON
                        "  'format' = 'json',\n" +
                        // 配置当缺少字段时不会抛出异常
                        "  'json.fail-on-missing-field' = 'false',\n" +
                        // 配置忽略JSON解析错误
                        "  'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );



    }

    private static TableEnvironment getTableEnv() {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner().inStreamingMode().build();

// 根据配置的环境设置创建TableEnvironment实例
        TableEnvironment tabEnv = TableEnvironment.create(settings);

// 获取并配置TableEnvironment的配置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
// 设置本地时间区域为Asia/Shanghai，以便时间处理操作使用正确的时区
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
// 设置默认并行度为1，控制执行的并行任务数
        configuration.setString("table.exec.resource.default-parallelism", "1");
// 设置状态生存时间（TTL）为5秒，自动清理过期的状态数据
        configuration.setString("table.exec.state.ttl", "5 s");

// 返回配置好的TableEnvironment实例
        return tabEnv;
    }











}
